package com.intct.util;

import com.intct.common.Constant;
import com.ververica.cdc.connectors.mysql.source.MySqlSource;
import com.ververica.cdc.connectors.mysql.table.StartupOptions;
import com.ververica.cdc.debezium.JsonDebeziumDeserializationSchema;

/**
 * @author gufg
 * @since 2025-07-03 09:22
 */
public class MysqlUtil {
    /**
     * 配置FLinkCDC中MySQL数据源
     * @param startupOptions
     * @return
     */
    public static MySqlSource<String> getMysqlSource(StartupOptions startupOptions, String dbList, String tList) {
        return MySqlSource.<String>builder()
                .hostname(Constant.MYSQL_HOSTNAME)
                .port(Constant.MYSQL_PORT)
                .username(Constant.MYSQL_USERNAME)
                .password(Constant.MYSQL_PASSWORD)
                .databaseList(dbList)
                .tableList(tList)
                .deserializer(new JsonDebeziumDeserializationSchema())
                .startupOptions(startupOptions)
                .build();
    }

    /**
     * 配置FLinkCDC中MySQL数据源
     * @param startupOptions
     * @return
     */
    public static MySqlSource<String> getMysqlSource(StartupOptions startupOptions) {
        return MySqlSource.<String>builder()
                .hostname(Constant.MYSQL_HOSTNAME)
                .port(Constant.MYSQL_PORT)
                .username(Constant.MYSQL_USERNAME)
                .password(Constant.MYSQL_PASSWORD)
                .databaseList(Constant.FLINK_CDC_DATABASE_LIST)
                .tableList(Constant.FLINK_CDC_TABLE_LIST)
                .deserializer(new JsonDebeziumDeserializationSchema())
                .startupOptions(startupOptions)
                .build();
    }
}
